# -*- coding: utf-8 -*-

import rfc822
from datetime import datetime, timedelta, date

from stream import models, utils
from stream.models import FeedChannel, FeedEntry, BookmarkEntry, EntryBase
import stream.feeds as feeds

from feedgenerator import FeedGenerator, EntryGenerator
import feedprovider
import testcase
import base

class FeedParsingTest(testcase.TestCase):
    def setUp(self):
        self.tag_list = ['tag1', 'tag2']
        self.entry_generator = EntryGenerator(tags = self.tag_list)
        self.feed_generator = FeedGenerator(channel_items = [self.entry_generator.createEntryText()])

        feeds.find_feed.mock(utils.identity)

    def testParsingAndCreation(self):
        """
        Parses a feed from string and puts it into the db.
        """
        now_time = datetime(2009, 9, 10, 0, 0, 0)
        models.now.mockValue(now_time)
        feeds.retrieve_feed.mockValue(feeds.feedparser.parse(self.feed_generator.default_feed))

        feed = FeedChannel.get_or_create(url = "xxx")
        self.assertEqual(feed.title, self.feed_generator.title)
        self.assertEqual(feed.last_updated, now_time)
        self.assertEqual(feed.entries.count(), 1)

        entry = feed.entries.all()[0]
        # TODO: relying on using default EntryGenerator in FeedGenerator. Should add a reference to FeedGenerator.
        self.assertEqual(entry.guid, self.entry_generator.id)
        self.assertEqual(entry.title, self.entry_generator.title)
        self.assertEqual(entry.link, self.entry_generator.link)
        self.assertEqual(entry.description, self.entry_generator.description)
        self.assertEqual(entry.updated, feeds.convertDate(rfc822.parsedate(self.entry_generator.pub_date)))
        self.assertTrue(utils.unorderedEqual(self.tag_list, map(lambda x: x.name, entry.tags.all())))

    def testUpdating(self):
        now_time = datetime(2009, 9, 10, 0, 0, 0)
        models.now.mockValue(now_time)

        feeds.retrieve_feed.mockValue(feeds.feedparser.parse(self.feed_generator.default_feed))
        feed = FeedChannel.get_or_create(url = "xxx")
        self.assertEqual(FeedChannel.objects.filter(title = self.feed_generator.title)[0].entries.count(), 1)

        # The entry has been updated
        old_description = self.entry_generator.description
        self.entry_generator.description = "Some new shit"
        self.entry_generator.pub_date = "Sat, 19 Sep 2009 0:00:01 GMT"
        self.entry_generator.tags = ['new_tag']
        self.assertNotEqual(self.entry_generator.description, old_description)
        constructed_entry = self.entry_generator.createEntryText()

        self.feed_generator.channel_items = [constructed_entry]
        constructed_feed = self.feed_generator.createFeedText()
        now_time = datetime(2009, 9, 20, 0, 0, 0)

        feeds.retrieve_feed.mockValue(feeds.feedparser.parse(constructed_feed))
        FeedChannel.objects.get(title = self.feed_generator.title).update()

        query = FeedChannel.objects.filter(title = self.feed_generator.title)
        self.assertEqual(query.count(), 1)
        self.assertEqual(query[0].entries.count(), 1) # Updated, not added
        entry = query[0].entries.all()[0]
        self.assertEqual(entry.guid, self.entry_generator.id)
        self.assertEqual(entry.description, self.entry_generator.description)
        self.assertTrue(utils.unorderedEqual(self.entry_generator.tags, map(lambda x: x.name, entry.tags.all())))

    def testNoGuid(self):
        '''
        Test parsing a feed with no guid and publishing date.
        '''
        feed_title = 'FeedTitle'
        entry_title1 = 'EntryTitle1'
        entry_title2 = 'EntryTitle2'
        entry_link1 = 'http://1.example.com'
        entry_link2 = 'http://2.example.com'
        feed_text = u"""<?xml version="1.0" encoding="windows-1251"?>
        <rss version="2.0">
        <channel>
           <title>%s</title>
           <link>http://lenta.ru/</link>
           <item>
              <title>%s</title>
              <link>%s</link>
              <description>EntryDescription</description>
           </item>
           <item>
              <title>%s</title>
              <link>%s</link>
              <description>EntryDescription</description>
              <pubDate>Sun, 03 Jan 2010 14:36:57 +0300</pubDate>
           </item>
        </channel>
        </rss>
        """ % (feed_title, entry_title1, entry_link1, entry_title2, entry_link2)
        feeds.retrieve_feed.mockValue(feeds.feedparser.parse(feed_text))
        feed = FeedChannel.get_or_create(url = 'xxx')
        self.assertEqual(feed.title, feed_title)
        self.assertEqual(feed.entries.count(), 2)
        entries = feed.entries.all()
        first_entry = feed.entries.filter(title = entry_title1)[0]
        self.assertEqual(first_entry.link, entry_link1)
        self.assertEqual(first_entry.created, FeedChannel.DEFAULT_DATE)
        second_entry = feed.entries.filter(title = entry_title2)[0]
        self.assertEqual(second_entry.link, entry_link2)
        self.assertEqual(second_entry.created.date(), date(2010, 01, 03))

    def testSpaceCategory(self):
        category_original = "my category"
        category_expected = "my_category"
        feed_text = """<?xml version="1.0" encoding="windows-1251"?>
        <rss version="2.0">
        <channel>
           <title>Title</title>
           <link>http://lenta.ru/</link>
           <item>
              <title>Title</title>
              <link>http://link</link>
              <description>EntryDescription</description>
              <category>%s</category>
           </item>
        </channel>
        </rss>""" % category_original
        feeds.retrieve_feed.mockValue(feeds.feedparser.parse(feed_text))
        feed = FeedChannel.get_or_create(url = 'xxx')
        entry = feed.entries.all()[0]
        self.assertEqual(entry.tags.all()[0].name, category_expected)

    def tearDown(self):
        feeds.retrieve_feed.reset()
        feeds.find_feed.reset()

class NonExistentTest(testcase.TestCase):
    def testNonExistentFeed(self):
        #A discoverable link with non-existent feed
        link = 'Does not exist'
        self.assertTrue(feeds.find_feed(link) is not None)
        self.assertRaises(FeedChannel.NotFoundError, FeedChannel.get_or_create, url = link)
        self.assertEqual(FeedChannel.objects.count(), 0)

        #A non-discoverable link
        feeds.find_feed.mockValue(None)
        self.assertRaises(FeedChannel.NotFoundError, FeedChannel.get_or_create, url = link)
        self.assertEqual(FeedChannel.objects.count(), 0)
        feeds.find_feed.reset()

class DeliciousTest(testcase.TestCase):
    def setUp(self):
        self.provider = feedprovider.FeedProvider(data_dict = feedprovider.delicious_data.copy())
        models.DeliciousChannel.retrieveUrl.mock(self.provider.retrieve_feed)
        self.now_time = datetime(2009, 9, 10, 0, 0, 0)
        models.now.mockValue(self.now_time)

    def testParsing(self):
        channel = models.DeliciousChannel.objects.create(author = 'command0')
        channel.update()
        self.assertEqual(channel.last_updated, self.now_time)
        self.assertEqual(channel.entries.count(), 10)
        first_entry = channel.entries.filter(link = 'http://delicious.com/help/feeds')[0]
        self.assertTagListEqual(['delicious', 'feeds', 'diploma'], first_entry.tags.all())

    def testUpdating(self):
        channel = models.DeliciousChannel.objects.create(author = 'command0')
        channel.update()
        self.assertEqual(channel.last_updated, self.now_time)
        self.now_time = datetime(2009, 10, 10, 0, 0, 0)
        models.now.mockValue(self.now_time)
        self.provider.data_dict['http://feeds.delicious.com/v2/json/command0?count=10'] = open('stream/testcases/command0_delicious_updated.json').read()
        channel.update()
        self.assertEqual(channel.last_updated, self.now_time)
        first_entry = channel.entries.filter(link = 'http://delicious.com/help/feeds')[0]
        self.assertTagListEqual(['delicious', 'diploma'], first_entry.tags.all())

    def testNonExisting(self):
        self.assertRaises(models.DeliciousChannel.NotFoundError, models.DeliciousChannel.get_or_create, author = 'nonexistent')
        self.assertEqual(models.DeliciousChannel.objects.count(), 0)

    def testLongTitle(self):
        long_title = 'start' + ''.zfill(300)
        delicious_data = '[{"u":"http://example.org", "d": "%s", "t" : ["django", "models"], "dt" : "2010-01-27T10:26:36Z", "n" : "", "a" : "command0"}]' % long_title
        self.provider.data_dict[feedprovider.delicious_url] = delicious_data
        channel = models.DeliciousChannel.objects.create(author = 'command0')
        channel.update()
        self.assertEqual(channel.entries.count(), 1)
        entry = channel.entries.all()[0]
        self.assertTrue(entry.title.startswith('start0'))

    def tearDown(self):
        models.DeliciousChannel.retrieveUrl.reset()
        super(DeliciousTest, self).tearDown()

class TwitterTest(testcase.TestCase):
    def setUp(self):
        self.provider = feedprovider.FeedProvider(data_dict = feedprovider.twitter_data)
        models.TwitterChannel.retrieveUrl.mock(self.provider.retrieve_feed)
        self.now_time = datetime(2009, 9, 10, 0, 0, 0)
        models.now.mockValue(self.now_time)

    def testParsing(self):
        channel = models.TwitterChannel.objects.create(author = 'command000')
        channel.update()
        self.assertEqual(channel.last_updated, self.now_time)
        self.assertEqual(channel.entries.count(), 1)
        entry = channel.entries.all()[0]
        self.assertEqual(entry.link, 'http://twitter.com/command000/statuses/4343309143')
        self.assertEqual(entry.twitter_id, '4343309143')
        self.assertEqual(entry.description, u'I just voted for "Provider logos with licensing\u2026" what do you think? http://uservoice.com/a/VAoU9 #OpenID #feedback')
        self.assertTrue(utils.unorderedEqual(['OpenID', 'feedback'], map(lambda x: x.name, entry.tags.all())))

    def testNonExisting(self):
        self.assertRaises(models.TwitterChannel.NotFoundError, models.TwitterChannel.get_or_create, author = 'nonexistent')
        self.assertEqual(models.TwitterChannel.objects.count(), 0)

    def tearDown(self):
        models.TwitterChannel.retrieveUrl.reset()
        super(TwitterTest, self).tearDown()

from django.test import TransactionTestCase
class TransactionTest(TransactionTestCase):
    '''
    Test transaction maanagement.
    '''
    # Hack: Depends on implementation of FeedChannel
    class DbFailer:
        def fail(self):
            '''
            Sets the 'called' attribute to True and executes a bad SQL.
            '''
            from django.db import connection
            cursor = connection.cursor()
            self.called = True
            cursor.execute('select;')

        bozo = property(fail)

    def testOPMLInterrupt(self):
        bad_url = 'http://example.org'
        opml_file = """<?xml version="1.0" encoding="UTF-8"?>
        <opml version="1.0">
            <body>
                <outline text="My blog" title="My blog" type="rss" xmlUrl="%s"/>
                <outline text="Bad link" title="Bad link" type="rss" xmlUrl="%s"/>
            </body>
        </opml>""" % (feedprovider.lj_url, bad_url)

        data_dict = feedprovider.lj_data.copy()
        failer = TransactionTest.DbFailer()
        data_dict.update({ bad_url : failer })
        provider = feedprovider.FeedProvider(data_dict = data_dict)
        feeds.find_feed.mock(utils.identity)
        feeds.retrieve_feed.mock(lambda url: (provider.retrieve_feed(url)))

        from stream.forms import OpmlForm
        from django.core.files.uploadedfile import SimpleUploadedFile

        form = OpmlForm({ 'file' : 'opml', 'form_name' : 'opml_form' }, { 'file' : SimpleUploadedFile('name.xml', opml_file) })
        if not form.is_valid():
            self.fail()
        form.save(models.User.objects.create(username = 'a', password = 'b'))
        self.assertTrue(failer.called)

        self.assertEqual(models.FeedChannel.objects.count(), 1)
        channel = models.FeedChannel.objects.all()[0]
        self.assertEqual(channel.entries.count(), 12)

    def testUpdateInterrupt(self):
        '''
        Check that if a DB error happens, all previously updated channels will still contain the updates
        '''
        url1 = 'http://example.com/'
        url2 = 'http://example.org/'
        feed_generator = FeedGenerator()
        old_title = 'Old title'
        entry_generator = EntryGenerator(link = url1 + '1', id = 'id1', title = old_title)
        feed_generator.channel_items = [entry_generator.createEntryText()]
        feed1 = feeds.feedparser.parse(feed_generator.createFeedText())

        entry_generator.link = url2 + '1'
        entry_generator.id = 'id2'
        feed_generator.channel_items = [entry_generator.createEntryText()]
        feed2 = feeds.feedparser.parse(feed_generator.createFeedText())

        provider = feedprovider.FeedProvider(data_dict = { url1 : feed1, url2 : feed2 })
        feeds.find_feed.mock(utils.identity)
        feeds.retrieve_feed.mock(lambda url, **kwargs : provider.retrieve_feed(url))

        for url in (url1, url2):
            models.FeedChannel.get_or_create(url = url)

        new_title = 'New title'
        entry_generator.link = url1 + '1'
        entry_generator.id = 'id1'
        entry_generator.title = new_title
        feed_generator.channel_items = [entry_generator.createEntryText()]
        feed1_updated = feeds.feedparser.parse(feed_generator.createFeedText())

        failer = TransactionTest.DbFailer()
        provider.data_dict = { url1 : feed1_updated, url2 : failer }

        from stream import updater
        updater.updateStaleChannels(0, True)
        self.assertTrue(failer.called)

        updated_entry = models.FeedEntry.objects.get(guid = 'id1')
        self.assertEqual(updated_entry.title, new_title)

    def tearDown(self):
        feeds.retrieve_feed.reset()
        super(TransactionTest, self).tearDown()
